package com.wmh.baseservice.common.redis.mq.core.stream;

import cn.hutool.core.collection.CollUtil;
import com.alibaba.fastjson2.JSONObject;
import com.google.common.collect.HashMultimap;
import com.google.common.collect.Multimap;

import java.util.Collection;
import java.util.Comparator;

/**
 * redis stream mq 消息失败处理器工厂
 *
 * @author 审心
 */
public class RedisStreamFailHandlerFactory {

    /**
     * 失败处理器工厂 存储字段
     * key: 消息key
     * value: {@link RedisStreamFailHandler} 实现类
     */
    private final Multimap<String, RedisStreamFailHandler<? extends AbstractRedisStreamMessage>>
            failHandlerMultimap = HashMultimap.create();

    public void put(String streamKey, RedisStreamFailHandler<? extends AbstractRedisStreamMessage> handler) {
        this.failHandlerMultimap.put(streamKey, handler);
    }

    /**
     * 获取排序后
     *
     * @param k 消息体类对象
     * @return 对应的消息拦截器
     */
    public Collection<RedisStreamFailHandler<? extends AbstractRedisStreamMessage>> getAndSort(String k) {
        Collection<RedisStreamFailHandler<? extends AbstractRedisStreamMessage>> failHandlers = failHandlerMultimap.get(k);
        CollUtil.sort(failHandlers, Comparator.comparingInt(RedisStreamFailHandler::getPriority));
        return failHandlers;
    }

    /**
     * 调用 key 对应的 handler
     *
     * @param streamKey key
     * @param json      消息体的 JSONObject 类型
     */
    public void handler(String streamKey, JSONObject json) {
        Collection<RedisStreamFailHandler<? extends AbstractRedisStreamMessage>> handlers = this.getAndSort(streamKey);
        handlers.forEach(handler -> handler.handler(json));
    }
}
